package com.example.actordemo3.pubSub;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.cluster.pubsub.DistributedPubSub;
import akka.cluster.pubsub.DistributedPubSubMediator;
import akka.event.Logging;
import akka.event.LoggingAdapter;

public class pubSubTets {

    static class Subscriber extends AbstractActor {
        LoggingAdapter log = Logging.getLogger(getContext().system(), this);

        public Subscriber() {
            ActorRef mediator = DistributedPubSub.get(getContext().system()).mediator();
            // subscribe to the topic named "content"
            mediator.tell(new DistributedPubSubMediator.Subscribe("content", getSelf()), getSelf());
        }

        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(String.class, msg -> log.info("Got: {}", msg))
                    .match(DistributedPubSubMediator.SubscribeAck.class, msg -> log.info("subscribed"))
                    .build();
        }
    }

    static class Publisher extends AbstractActor {

        // activate the extension
        ActorRef mediator = DistributedPubSub.get(getContext().system()).mediator();

        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(
                            String.class,
                            in -> {
                                String out = in.toUpperCase();
                                mediator.tell(new DistributedPubSubMediator.Publish("content", out), getSelf());
                            })
                    .build();
        }
    }

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("sys1");
        system.actorOf(Props.create(Subscriber.class), "subscriber1");
// another node
        system.actorOf(Props.create(Subscriber.class), "subscriber2");
        system.actorOf(Props.create(Subscriber.class), "subscriber3");
        ActorRef publisher = system.actorOf(Props.create(Publisher.class), "publisher");
// after a while the subscriptions are replicated
        publisher.tell("hello", null);
    }




}
